package org.example.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;

/**
 * RocketMQ事务的3个状态:
 * TransactionStatus.CommitTransaction：提交事务消息，消费者可以消费此消息
 * TransactionStatus.RollbackTransaction：回滚事务，它代表该消息将被删除，不允许被消费。
 * TransactionStatus.Unknown ：中间状态，它代表需要检查消息队列来确定状态。
 *
 * @author : Charles
 * @date : 2021/6/10
 */
@Slf4j
@Service
// @RocketMQTransactionListener表明这个一个生产端的消息监听器，需要配置监听的事务消息生产者组。
// 实现RocketMQLocalTransactionListener接口，重写执行本地事务的方法和检查本地事务方法
@RocketMQTransactionListener
public class TxProducerListener implements RocketMQLocalTransactionListener {

    /**
     * 每次推送消息会执行executeLocalTransaction方法，首先会发送半消息，到这里的时候是执行具体本地业务，
     * 执行成功后手动返回RocketMQLocalTransactionState.COMMIT状态，
     * 这里是保证本地事务执行成功，如果本地事务执行失败则可以返回ROLLBACK进行消息回滚。 此时消息只是被保存到broker，并没有发送到topic中，broker会根据本地返回的状态来决定消息的处理方式。
     * @param message
     * @param arg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object arg) {
        log.info("开始执行本地事务");
        RocketMQLocalTransactionState state;
        try{
            Integer i = (Integer) arg;
            if (i % 2 == 0) {
                // 偶数抛出异常
                int a = i / 0;
            }
            // COMMIT:即生产者通知Rocket该消息可以消费
            state = RocketMQLocalTransactionState.COMMIT;
            log.info("本地事务已提交。{}",message.getHeaders().get("key").toString());
        }catch (Exception e){
            log.info("执行本地事务失败。{}",e);
            // ROLLBACK:即生产者通知Rocket将该消息删除
            state = RocketMQLocalTransactionState.ROLLBACK;
        }
        return state;
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        log.info("开始执行回查");
        // 判断具体业务逻辑，来决定是否回滚还是提交
        boolean flag = false;
        if ( flag ) {
            log.info("回滚半消息");
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        log.info("提交半消息");
        return RocketMQLocalTransactionState.COMMIT;
    }
}

